package vn.com.epi.re.tools;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;

import vn.com.epi.re.ApplicationContext;
import vn.com.epi.re.table.ArticleTable;
import vn.com.epi.re.table.UserTable;

public class Cleaner extends Configured implements Tool {
    
    private final Log log = LogFactory.getLog(Cleaner.class);

    private static final String REPORT_FILE = "/data/recommendation/baomoilogs/report.txt";
//    private static final String REPORT_FILE = "/home/trung/Test/script/report.txt";

    public static final long TWO_DAYS = 172800; 
    public static final long TIME_LIMIT = 432000; // 5 days
    public static final int GROUP_SIZE = 1000;
    
    private int numberOfUsers = 0;
    private int numberOfArticles = 0;

    
    public void deleteOldVisitedArticles() throws IOException {
        log.info("----------------* start delete old visited article *------------------");
        HTableInterface table = ApplicationContext.get().getTable(UserTable.NAME);
        try {
            List<Put> puts = Lists.newArrayList();
            int counter = 0;
            Scan scan = new Scan();
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES);        
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS);        
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_READING_PREFERENCES);        
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_LAST_VISITED_TIMESTAMP);        
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_TOTAL_VISITED_COUNT);        
            ResultScanner rs = table.getScanner(scan);
            long current = System.currentTimeMillis() / 1000L;
            for (Result r:rs) {
                long maxTimestamp = 0;
                int totalVisit = 0;
                counter++;
                StringBuilder articleSb = new StringBuilder();
                StringBuilder preferenceSb = new StringBuilder();
                StringBuilder timestampSb = new StringBuilder();
                String timestampCol = Bytes.toString(r.getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS));
                if (!Strings.isNullOrEmpty(timestampCol)) {
                    String articleCol = Bytes.toString(r.getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES));
                    String preferenceCol = Bytes.toString(r.getValue(UserTable.FAM_MAIN, UserTable.COL_READING_PREFERENCES));
                    String timestamps[] = timestampCol.split(",");
                    String[] articles = articleCol.split(",");
                    String[] preferences = preferenceCol.split(",");
                    for (int i = 0; i < articles.length; i++) {
                        long timestamp = Long.parseLong(timestamps[i]);
                        maxTimestamp = Math.max(maxTimestamp, timestamp);
                        if (current - timestamp < TWO_DAYS) {
                            totalVisit++;
                            articleSb.append(articles[i]).append(",");
                            preferenceSb.append(preferences[i]).append(",");
                            timestampSb.append(timestamps[i]).append(",");
                        }
                    }
                    StringUtils.deleteLastChar(articleSb);
                    StringUtils.deleteLastChar(preferenceSb);
                    StringUtils.deleteLastChar(timestampSb);
                }                
                Put put = new Put(r.getRow());
                put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES, Bytes.toBytes(articleSb.toString()));
                put.add(UserTable.FAM_MAIN, UserTable.COL_READING_PREFERENCES, Bytes.toBytes(preferenceSb.toString()));
                put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS, Bytes.toBytes(timestampSb.toString()));
                put.add(UserTable.FAM_MAIN, UserTable.COL_LAST_VISITED_TIMESTAMP, Bytes.toBytes(maxTimestamp));
                put.add(UserTable.FAM_MAIN, UserTable.COL_TOTAL_VISITED_COUNT, Bytes.toBytes(totalVisit));
                puts.add(put);
                if (counter == GROUP_SIZE) {                    
                    counter = 0;
                    table.put(puts);
                    puts = Lists.newArrayList();    
                }
            }
            if (counter > 0)
                table.put(puts);
            log.info("----------------* completed delete articles *------------------");            
        } catch (Exception e) {
          // ignore
            System.out.println("Error while delete old articles: " + e.toString());
            log.info("Error while delete old articles !!!!");
        } finally {
            TableUtils.closeQuietly(table);
        } 
    }
    
    public void deleteNoiseUsers() throws IOException {
        log.info("----------------* start delete noise users *------------------");
        HTableInterface userTable = ApplicationContext.get().getTable(UserTable.NAME);
        int deletedUsers = 0;
        try {
            List<Delete> deletes = Lists.newArrayList();
            Scan scan = new Scan();
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_LAST_VISITED_TIMESTAMP);        
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_TOTAL_VISITED_COUNT);        
            ResultScanner rs = userTable.getScanner(scan);
            long current = System.currentTimeMillis() / 1000L;
            int counter = 0;
            for (Result r:rs) {
                numberOfUsers++;
                long lastVisitedTimestamp = Bytes.toLong(r.getValue(UserTable.FAM_MAIN, UserTable.COL_LAST_VISITED_TIMESTAMP));
                int totalVisited = Bytes.toInt(r.getValue(UserTable.FAM_MAIN, UserTable.COL_TOTAL_VISITED_COUNT));
                if (current - lastVisitedTimestamp >= TWO_DAYS) {
                    if (150 < totalVisited || totalVisited < 20) {
                        Delete delete = new Delete(r.getRow());
                        deletes.add(delete);
                        counter++;
                        deletedUsers++;
                        numberOfUsers--;
                    }                        
                }
                if (counter == GROUP_SIZE) {
                    userTable.delete(deletes);
                    deletes = Lists.newArrayList();
                    counter = 0;
                }
            }
            if (counter > 0) {
                userTable.delete(deletes);
            }
            log.info("----------------* completed delete users *------------------");
            System.out.println("Deleted " + deletedUsers + " users");
        } catch (Exception e) {
          // ignore
            System.out.println("Error while delete users: " + e.toString());
            log.info("Error while delete users !!!!");
        } finally {
            TableUtils.closeQuietly(userTable);
        }         
    }
    
    public void deleteOldArticles() throws IOException {
        log.info("----------------* start delete old article *------------------");
        HTableInterface articleTable = ApplicationContext.get().getTable(ArticleTable.NAME);
        HTableInterface recommendationTable = ApplicationContext.get().getTable(ArticleTable.NAME);
        int deletedArticles = 0;
        try {
            List<Delete> deletes = Lists.newArrayList();
            Scan scan = new Scan();
            scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS);        
            ResultScanner rs = articleTable.getScanner(scan);
            long current = System.currentTimeMillis() / 1000L;
            int counter = 0;
            for (Result r:rs) {
                numberOfArticles++;
                long firstVisitedTimestamp = Bytes.toLong(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS));
                if (current - firstVisitedTimestamp > TIME_LIMIT) {
                    Delete delete = new Delete(r.getRow());
                    deletes.add(delete);
                    counter++;
                    deletedArticles++;
                    numberOfArticles--;
                }
                if (counter == GROUP_SIZE) {
                    articleTable.delete(deletes);
                    recommendationTable.delete(deletes);
                    deletes = Lists.newArrayList();
                    counter = 0;
                }
            }
            if (counter > 0) {
                articleTable.delete(deletes);
                recommendationTable.delete(deletes);
            }
            log.info("----------------* completed delete articles *------------------");            
            System.out.println("Deleted " + deletedArticles + " articles");
        } catch (Exception e) {
          // ignore
            System.out.println("Error while delete article: " + e.toString());
            log.info("Error while delete article !!!!");
        } finally {
            TableUtils.closeQuietly(articleTable);
        }         
    }    

    private void report() throws IOException {
        BufferedWriter bw = null;
        try {
            File file = new File(REPORT_FILE);
            if (!file.exists()) {
                file.createNewFile();
            }
            FileWriter fw = new FileWriter(file.getAbsoluteFile());
            bw = new BufferedWriter(fw);
            bw.write("User:" + numberOfUsers + "\n");
            bw.write("Article:" + numberOfArticles);
        } catch (Exception e) {
            //ignore
        } finally {
            if (bw != null)
                try {
                    bw.close();
                } catch (Exception ex) {                    
                }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        deleteOldVisitedArticles();
        deleteNoiseUsers();
        deleteOldArticles();
        report();
        return 0;
    }
    
    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ApplicationContext c = ApplicationContext.init();
        try {
            ToolRunner.run(new Cleaner(), args);
        } finally {
            Closeables.closeQuietly(c);            
        }        
        long endTime = System.currentTimeMillis();
        System.out.println("total time used: " + (endTime - startTime) + " ms.");        
    }
    
}
